package com.hrt.kuducode

import java.util

import org.apache.kudu.client.CreateTableOptions
import org.apache.kudu.spark.kudu.KuduContext
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import org.apache.spark.sql.types.{DataTypes, StructField, StructType}


/**
 * SparkCore操作kudu
 */
case class PersonInfo(id:Int,name:String,age:Int,score:Double)
object SparkKudu {

  def main(args: Array[String]): Unit = {
    val session: SparkSession = SparkSession.builder().master("local").appName("test").getOrCreate()

    val sc: SparkContext = session.sparkContext
    sc.setLogLevel("Error")

    //1) 创建kuduContext
    val kuduContext = new KuduContext("cm1:7051,cm2:7051,cm3:7051",sc)

//    val schema = StructType(List[StructField](
//      StructField("id", DataTypes.IntegerType, false),
//      StructField("name", DataTypes.StringType, false),
//      StructField("age", DataTypes.IntegerType, false),
//      StructField("score", DataTypes.DoubleType, false)
//    ))

    // 2) 创建Kudu 表
//    val options = new CreateTableOptions()
//    options.addHashPartitions(util.Arrays.asList("id"),5)
//    kuduContext.createTable("spark_t1",schema,Seq[String]("id"),options)


    // 3)插入数据
//    val list = List[PersonInfo](
//      PersonInfo(1,"zs",18,100),
//      PersonInfo(2,"ls",19,200),
//      PersonInfo(3,"ww",20,300)
//    )
//
//    import session.implicits._
//    val frame: DataFrame = list.toDF()
//
//    kuduContext.insertRows(frame,"spark_t1")

    //4 ) 查询数据
//    val result: RDD[Row] = kuduContext.kuduRDD(sc,"spark_t1",Seq[String]("id","name","age","score"))
//    result.foreach(println)

    // 5) 更新数据

//    val list = List[PersonInfo](
//      PersonInfo(1,"sss",1,2)
//    )
//    import session.implicits._
//    val frame: DataFrame = list.toDF()
//    kuduContext.upsertRows(frame,"spark_t1")
//
//    val result: RDD[Row] = kuduContext.kuduRDD(sc,"spark_t1",Seq[String]("id","name","age","score"))
//    result.foreach(println)


    //6） 删除数据
    val list = List[PersonInfo](
      PersonInfo(2,"sss",1,2)
    )
    import session.implicits._
    val frame: DataFrame = list.toDF()
    kuduContext.deleteRows(frame.select("id"),"spark_t1")

    val result: RDD[Row] = kuduContext.kuduRDD(sc,"spark_t1",Seq[String]("id","name","age","score"))
    result.foreach(println)

  }
}
